package com.team5101.client;


import com.alibaba.fastjson.JSON;
import com.team5101.client.handler.DemoClientHandler;

import com.team5101.configEntity.Constant;
import com.team5101.configEntity.RpcProperties;
import com.team5101.configEntity.ZooKeeperProperties;
import com.team5101.encode.RpcEncoder;
import com.team5101.model.RpcRequest;
import com.team5101.serializer.JSONSerializer;
import com.team5101.util.Obj2ByteUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;


@Component
public class RpcConsumer {

    private Logger logger = LoggerFactory.getLogger(RpcConsumer.class);

    @Autowired
    private RpcProperties rpcProperties;

//	private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private DemoClientHandler demoClientHandler;

    private Bootstrap bootstrap;
    private EventLoopGroup eventExecutors = new NioEventLoopGroup();


    /**
     * 每个服务端（不同的服务节点）都需要建立一个连接
     */
    private static Map<String, Channel> channelMap = new ConcurrentHashMap<>();


    /**
     * 提供服务的父节点
     */
    private String producerZnodeName = "/" + Constant.PRODUCER_TYPE;

    /**
     * 负载均衡的父节点
     */
    private String loadBalancePath = "/" + Constant.LOADBALANCE;

    @Autowired
    private CuratorFramework curatorFramework;

    @Autowired
    private ZooKeeperProperties zkProperties;


    public void startUp() throws Exception {
        //连接zk服务端
        this.curatorFramework.start();
        this.curatorFramework.blockUntilConnected();
        logger.info("服务器提供端：{} ，已连接上zookeeper服务端：{}", rpcProperties.getProducerName(), zkProperties.getZkHost() + ":" + zkProperties.getPort());

        //初始化Netty客户端
        initClient();

        //获取所有的子节点，每个子节点就是一个Netty服务端，需要建立一个连接
        List<String> children = getChildren();
        //创建socket连接
        //2.客户端启动时，从Zookeeper中获取所有服务提供端节点信息，客户端与每一个服务端都建立连接
        createChannel(children);

        //监听服务
        //3.服务端重新上线，客户端能感知到，并且与重新上线的服务端重新建立连接
        //4.某个服务端下线后，Zookeeper注册列表会自动剔除下线的服务端节点，客户端与下线的服务端断开连接
        listenersOfChild();

    }

    /**
     * 服务端重新上线，客户端能感知到，并且与重新上线的服务端重新建立连接
     * 某个服务端下线后，Zookeeper注册列表会自动剔除下线的服务端节点，客户端与下线的服务端断开连接
     * <p>
     * 监听子节点
     *
     * @throws Exception
     */
    private void listenersOfChild() throws Exception {

        PathChildrenCache childrenCache = new PathChildrenCache(this.curatorFramework, this.producerZnodeName, true);
        PathChildrenCacheListener childrenCacheListener = (client, event) -> {
            logger.info("事件：{}", JSON.toJSONString(event));
            ChildData data = event.getData();
            switch (event.getType()) {
                case CHILD_ADDED://添加子节点
                    String znodeData = new String(data.getData());
                    String[] ipPort = znodeData.split(":");
                    logger.info("zk服务端新增服务 -- 当前子节点路径：{} ，当前子节点的数据：{}", data.getPath(), znodeData);
                    if (!channelMap.containsKey(data.getPath())) {
                        logger.info("~~~~~~~~~~~~~~~~ 创建连接：{}", data.getPath());
                        channelMap.put(data.getPath(), this.bootstrap.connect(ipPort[0], Integer.parseInt(ipPort[1])).sync().channel());
                    }
                    break;
                case CHILD_REMOVED://移除子节点
                    String znodeDatb = new String(data.getData());
                    String[] ipPortb = znodeDatb.split(":");
                    logger.info("zk服务端移除服务 -- 当前子节点路径：{} ，当前子节点的数据：{}", data.getPath(), znodeDatb);
                    if (channelMap.containsKey(data.getPath())) {
                        logger.info("~~~~~~~~~~~~~~~~ 移除连接：{}", data.getPath());
                        Channel channel = channelMap.get(data.getPath());
                        channel.close();
                        channelMap.remove(data.getPath());
                    }
                    break;
                case CHILD_UPDATED://子节点的数据变化
                    String znodeDataa = new String(data.getData());
                    String[] ipPorta = znodeDataa.split(":");
                    logger.info("zk服务端 服务数据变化 -- 当前子节点路径：{} ，当前子节点的数据：{}", data.getPath(), znodeDataa);
                    if (channelMap.containsKey(data.getPath())) {
                        logger.info("~~~~~~~~~~~~~~~~ {} 连接的数据发生变化", data.getPath());
                        Channel channel = channelMap.get(data.getPath());
                        channel.close();
                        channelMap.put(data.getPath(), this.bootstrap.connect(ipPorta[0], Integer.parseInt(ipPorta[1])).sync().channel());
                    }
                    break;
                default:
                    break;
            }
        };
        childrenCache.getListenable().addListener(childrenCacheListener);
        childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
    }

    /**
     * 客户端启动时，从Zookeeper中获取所有服务提供端节点信息，客户端与每一个服务端都建立连接
     * <p>
     * 与每个服务端建立连接
     *
     * @param providers 服务端在zk服务器上的节点路径
     */
    private void createChannel(List<String> providers) {

        providers.forEach(producerNode -> {
            if (!channelMap.containsKey(producerNode)) {

                String producerNodeData = null;
                try {
                    Stat stat = new Stat();
                    byte[] znodeData = this.curatorFramework.getData().storingStatIn(stat).forPath(producerNode);
                    if (znodeData != null && znodeData.length > 0) {
                        producerNodeData = new String(znodeData);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }

                if (Objects.nonNull(producerNodeData)) {
                    try {
                        String[] ipPort = producerNodeData.split(":");
                        logger.info("~~~~~~~~~~~~~~~~ 创建连接：{}", producerNode);
                        //连接
                        //bootstrap.connect(rpcProperties.getHost(), rpcProperties.getPort()).sync();
                        channelMap.put(producerNode, this.bootstrap.connect(ipPort[0], Integer.parseInt(ipPort[1])).sync().channel());
                    } catch (InterruptedException e) {
                        logger.error(e.getMessage(), e);
                    }
                }
            }
        });
    }


    public void initClient() throws InterruptedException {

        if (this.demoClientHandler == null) {
            this.demoClientHandler = new DemoClientHandler();
        }
//		this.accountClientHandler = new DemoClientHandler();

        //
//        EventLoopGroup eventExecutors = new NioEventLoopGroup();

//		Bootstrap bootstrap = new Bootstrap();

        if (this.bootstrap == null) {
            this.bootstrap = new Bootstrap();
        }

        bootstrap.group(eventExecutors)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();

                        //对RpcRequest编码
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        //对RpcResponse解码
                        //pipeline.addLast(new RpcDecoder(RpcResponse.class, new JSONSerializer()));
                        pipeline.addLast(new StringDecoder());

                        pipeline.addLast(demoClientHandler);
                    }
                });

        //连接
//        bootstrap.connect(rpcProperties.getHost(), rpcProperties.getPort()).sync();

    }


    //1.创建代理对象

    /**
     * @param serviceClass 被调用的服务类
     * @return
     */
    public Object createProxy(final Class<?> serviceClass) {
//		Object proxyInstance = Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{serviceClass}, (proxy, method, args) -> {

//			//2初始化客户端
//			if (this.accountClientHandler == null) {
//				initClient();
//			}
//
//			RpcRequest request = new RpcRequest();
//			request.setRequestId(UUID.randomUUID().toString());
//			request.setClassName(method.getDeclaringClass().getName());
//			request.setMethodName(method.getName());
//			request.setParameterTypes(method.getParameterTypes());
//			request.setParameters(args);
//
//
//			//3设置参数
//			this.accountClientHandler.setPara(request);
//
//			//请求服务端
//			return executor.submit(this.accountClientHandler).get();
//		});
//
//		return proxyInstance;
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{serviceClass}, (proxy, method, args) -> {

            RpcRequest request = new RpcRequest();
            String uuid = UUID.randomUUID().toString();
            request.setRequestId(uuid);
            request.setClassName(method.getDeclaringClass().getName());
            request.setMethodName(method.getName());
            request.setParameterTypes(method.getParameterTypes());
            request.setParameters(args);

            logger.info("requestId:{} ,请求服务器", uuid);
            sentMsg(request);
            Thread.sleep(1000);
            return getData(uuid);

        });
    }


    private void sentMsg(RpcRequest request) throws Exception {

        if (!channelMap.isEmpty()) {
            List<Channel> channels = new ArrayList<>(channelMap.values());

            //当客户端发起调用，每次都选择最后一次响应时间短的服务端进行服务调用，如果时间一致，随机选取一个服务端进行调用，从而实现负载均衡

            //  /NETTY-PRODUCER-1
            List<String> loadBalancePaths = this.curatorFramework.getChildren().forPath(this.loadBalancePath);
            //没有负载均衡节点
            if (loadBalancePaths.isEmpty()) {
                logger.info("{} 没有负载均衡的节点,随机选择一个", this.loadBalancePath);
                int size = channels.size();
                int i = new Random().nextInt(size);
                Channel channel = channels.get(i);
                channel.writeAndFlush(request).sync();
            } else {
                //说明最近只有一个被调用
                if (loadBalancePaths.size() == 1) {
                    logger.info("{} 负载均衡的节点只有一个：{}", this.loadBalancePath, loadBalancePaths.get(0));
                    Channel channel = channelMap.get(this.producerZnodeName + "/" + loadBalancePaths.get(0));
                    channel.writeAndFlush(request).sync();
                } else {
                    Map<Long, String> longMap = new TreeMap<>();
                    //最近都被调用过,选择最后一次响应时间短的服务端进行服务调用，如果时间一致，随机选取一个服务端进行调用，从而实现负载均衡
                    //  /ZK-RPC/LOAD-BALANCE/NETTY-PRODUCER-1
                    for (String balancePath : loadBalancePaths) {
                        String loadPath = this.loadBalancePath + "/" + balancePath;
                        Stat stat = new Stat();
                        byte[] loadPathTime = this.curatorFramework.getData().storingStatIn(stat).forPath(loadPath);
                        if (loadPathTime != null && loadPathTime.length > 0) {
                            long lastTime = Obj2ByteUtil.byteArrayToLong(loadPathTime);

                            //时间一样，就选用最后一个服务
                            longMap.put(lastTime, balancePath);
                        }
                    }

                    logger.info("{} 负载均衡的节点有多个：{}", this.loadBalancePath, longMap);
                    if (!longMap.isEmpty()) {
                        List<String> collect = new ArrayList<>(longMap.values());
                        String loadPath = collect.get(0);
                        Channel channel = channelMap.get(this.producerZnodeName + "/" + loadPath);
                        channel.writeAndFlush(request).sync();
                    }
                }
            }
        }
    }

    private List<String> getChildren() throws Exception {
        List<String> path = this.curatorFramework.getChildren().forPath(this.producerZnodeName);
        return path.stream().map(item -> this.joinPaths(item)).collect(Collectors.toList());
//		return path.stream().map(this::joinPaths).collect(Collectors.toList());

    }

    /**
     * 添加上的父节点
     *
     * @param childPath
     * @return
     */
    private String joinPaths(String childPath) {
        return this.producerZnodeName + "/" + childPath;
    }

    private static Map<String, Object> dataMap = new LinkedHashMap<String, Object>();

    private Object getData(String requestId) {
        logger.info("requestId:{} ,响应数据：{}", requestId, dataMap.get(requestId));
        return dataMap.get(requestId);
    }

    public static Map<String, Object> getDataMap() {
        return dataMap;
    }


    /**
     * 求Map<K,V>中Value(值)的最小值
     *
     * @param map
     * @return
     */
    public static Object getMinValue(Map<String, Long> map) {
        if (map == null)
            return null;
        Collection<Long> c = map.values();
        Object[] obj = c.toArray();
        Arrays.sort(obj);
        return obj[0];
    }

    /**
     * 求Map<K,V>中Value(值)的最大值
     *
     * @param map
     * @return
     */
    public static Object getMaxValue(Map<String, Long> map) {
        if (map == null)
            return null;
        int length = map.size();
        Collection<Long> c = map.values();
        Object[] obj = c.toArray();
        Arrays.sort(obj);
        return obj[length - 1];
    }

}
